package net.wicp.tams.common.binlog.parser;

import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.SocketTimeoutException;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collection;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * TODO: Document It!!
 * 
 * <pre>
 * DirectLogFetcher fetcher = new DirectLogFetcher();
 * fetcher.open(conn, file, 0, 13);
 * 
 * while (fetcher.fetch()) {
 * 	LogEvent event;
 * 	do {
 * 		event = decoder.decode(fetcher, context);
 * 
 * 		// process log event.
 * 	} while (event != null);
 * }
 * // connection closed.
 * </pre>
 * 
 * @author <a href="mailto:changyuan.lh@taobao.com">Changyuan.lh</a>
 * @version 1.0
 */
public final class DirectLogFetcher extends LogFetcher {

	protected static final Logger logger = LoggerFactory.getLogger(DirectLogFetcher.class);

	/** Command to dump binlog */
	public static final byte COM_BINLOG_DUMP = 18;

	/** Command to dump binlog */
	public static final byte COM_BINLOG_DUMP_GTID = 30;

	/** Packet header sizes */
	public static final int NET_HEADER_SIZE = 4;
	public static final int SQLSTATE_LENGTH = 5;

	/** Packet offsets */
	public static final int PACKET_LEN_OFFSET = 0;
	public static final int PACKET_SEQ_OFFSET = 3;

	/** Maximum packet length */
	public static final int MAX_PACKET_LENGTH = (256 * 256 * 256 - 1);

	/** BINLOG_DUMP options */
	public static final int BINLOG_DUMP_NON_BLOCK = 1;
	public static final int BINLOG_SEND_ANNOTATE_ROWS_EVENT = 2;

	private Connection conn;
	private OutputStream mysqlOutput;
	private InputStream mysqlInput;

	public DirectLogFetcher() {
		super(DEFAULT_INITIAL_CAPACITY, DEFAULT_GROWTH_FACTOR);
	}

	public DirectLogFetcher(final int initialCapacity) {
		super(initialCapacity, DEFAULT_GROWTH_FACTOR);
	}

	public DirectLogFetcher(final int initialCapacity, final float growthFactor) {
		super(initialCapacity, growthFactor);
	}

	private static final Object unwrapConnection(Object conn, Class<?> connClazz) throws IOException {
		while (!connClazz.isInstance(conn)) {
			try {
				Class<?> connProxy = Class.forName("org.springframework.jdbc.datasource.ConnectionProxy");
				if (connProxy.isInstance(conn)) {
					conn = invokeMethod(conn, connProxy, "getTargetConnection");
					continue;
				}
			} catch (ClassNotFoundException e) {
				// org.springframework.jdbc.datasource.ConnectionProxy not
				// found.
			}

			try {
				Class<?> connProxy = Class.forName("org.apache.commons.dbcp.DelegatingConnection");
				if (connProxy.isInstance(conn)) {
					conn = getDeclaredField(conn, connProxy, "_conn");
					continue;
				}
			} catch (ClassNotFoundException e) {
				// org.apache.commons.dbcp.DelegatingConnection not found.
			}

			try {
				if (conn instanceof java.sql.Wrapper) {
					Class<?> connIface = Class.forName("com.mysql.jdbc.Connection");
					conn = ((java.sql.Wrapper) conn).unwrap(connIface);
					continue;
				}
			} catch (ClassNotFoundException e) {
				// com.mysql.jdbc.Connection not found.
			} catch (SQLException e) {
				logger.warn("Unwrap " + conn.getClass().getName() + " to " + connClazz.getName() + " failed: "
						+ e.getMessage(), e);
			}

			return null;
		}
		return conn;
	}

	private static final Object invokeMethod(Object obj, Class<?> objClazz, String name) {
		try {
			Method method = objClazz.getMethod(name, (Class<?>[]) null);
			return method.invoke(obj, (Object[]) null);
		} catch (NoSuchMethodException e) {
			throw new IllegalArgumentException("No such method: \'" + name + "\' @ " + objClazz.getName(), e);
		} catch (IllegalAccessException e) {
			throw new IllegalArgumentException("Cannot invoke method: \'" + name + "\' @ " + objClazz.getName(), e);
		} catch (InvocationTargetException e) {
			throw new IllegalArgumentException("Invoke method failed: \'" + name + "\' @ " + objClazz.getName(),
					e.getTargetException());
		}
	}

	private static final Object getDeclaredField(Object obj, Class<?> objClazz, String name) {
		try {
			Field field = objClazz.getDeclaredField(name);
			field.setAccessible(true);
			return field.get(obj);
		} catch (NoSuchFieldException e) {
			throw new IllegalArgumentException("No such field: \'" + name + "\' @ " + objClazz.getName(), e);
		} catch (IllegalAccessException e) {
			throw new IllegalArgumentException("Cannot get field: \'" + name + "\' @ " + objClazz.getName(), e);
		}
	}

	/**
	 * Connect MySQL master to fetch binlog.
	 */
	public void open(Connection conn, String fileName, final int serverId) throws IOException {
		open(conn, fileName, BIN_LOG_HEADER_SIZE, serverId, false);
	}

	/**
	 * Connect MySQL master to fetch binlog.
	 */
	public void open(Connection conn, String fileName, final int serverId, boolean nonBlocking) throws IOException {
		open(conn, fileName, BIN_LOG_HEADER_SIZE, serverId, nonBlocking);
	}

	/**
	 * Connect MySQL master to fetch binlog.
	 */
	public void open(Connection conn, String fileName, final long filePosition, final int serverId) throws IOException {
		open(conn, fileName, filePosition, serverId, false);
	}

	/**
	 * Connect MySQL master to fetch binlog.
	 */
	public void open(Connection conn, String fileName, long filePosition, final int serverId, boolean nonBlocking)
			throws IOException {
		try {
			this.conn = conn;
			Class<?> connClazz = Class.forName("com.mysql.jdbc.ConnectionImpl");
			Object unwrapConn = unwrapConnection(conn, connClazz);
			if (unwrapConn == null) {
				throw new IOException(
						"Unable to unwrap " + conn.getClass().getName() + " to com.mysql.jdbc.ConnectionImpl");
			}

			// Get underlying IO streams for network communications.
			Object connIo = getDeclaredField(unwrapConn, connClazz, "io");
			if (connIo == null) {
				throw new IOException("Get null field:" + conn.getClass().getName() + "#io");
			}
			mysqlOutput = (OutputStream) getDeclaredField(connIo, connIo.getClass(), "mysqlOutput");
			mysqlInput = (InputStream) getDeclaredField(connIo, connIo.getClass(), "mysqlInput");

			if (filePosition == 0)
				filePosition = BIN_LOG_HEADER_SIZE;
			sendBinlogDump(fileName, filePosition, serverId, nonBlocking);
			position = 0;
		} catch (IOException e) {
			close(); /* Do cleanup */
			logger.error("Error on COM_BINLOG_DUMP: file = " + fileName + ", position = " + filePosition);
			throw e;
		} catch (ClassNotFoundException e) {
			close(); /* Do cleanup */
			throw new IOException("Unable to load com.mysql.jdbc.ConnectionImpl", e);
		}
	}

	public void openGtid(Connection conn, String gtidStr, final int serverId) throws IOException {
		openGtid(conn, gtidStr, serverId, false);
	}

	public void openGtid(Connection conn, String gtidStr, final int serverId, boolean nonBlocking) throws IOException {
		try {
			this.conn = conn;
			Class<?> connClazz = Class.forName("com.mysql.jdbc.ConnectionImpl");
			Object unwrapConn = unwrapConnection(conn, connClazz);
			if (unwrapConn == null) {
				throw new IOException(
						"Unable to unwrap " + conn.getClass().getName() + " to com.mysql.jdbc.ConnectionImpl");
			}

			// Get underlying IO streams for network communications.
			Object connIo = getDeclaredField(unwrapConn, connClazz, "io");
			if (connIo == null) {
				throw new IOException("Get null field:" + conn.getClass().getName() + "#io");
			}
			mysqlOutput = (OutputStream) getDeclaredField(connIo, connIo.getClass(), "mysqlOutput");
			mysqlInput = (InputStream) getDeclaredField(connIo, connIo.getClass(), "mysqlInput");
			sendBinlogDumpGtid(gtidStr, "", 4, serverId, nonBlocking);
			position = 0;
		} catch (IOException e) {
			close(); /* Do cleanup */
			logger.error("Error on COM_BINLOG_DUMP_GTID: gtidStr = " + gtidStr);
			throw e;
		} catch (ClassNotFoundException e) {
			close(); /* Do cleanup */
			throw new IOException("Unable to load com.mysql.jdbc.ConnectionImpl", e);
		}
	}

	/**
	 * Put a byte in the buffer.
	 * 
	 * @param b
	 *            the byte to put in the buffer
	 */
	protected final void putByte(byte b) {
		ensureCapacity(position + 1);

		buffer[position++] = b;
	}

	/**
	 * Put 16-bit integer in the buffer.
	 * 
	 * @param i16
	 *            the integer to put in the buffer
	 */
	protected final void putInt16(int i16) {
		ensureCapacity(position + 2);

		byte[] buf = buffer;
		buf[position++] = (byte) (i16 & 0xff);
		buf[position++] = (byte) (i16 >>> 8);
	}

	/**
	 * Put 32-bit integer in the buffer.
	 * 
	 * @param i32
	 *            the integer to put in the buffer
	 */
	protected final void putInt32(long i32) {
		ensureCapacity(position + 4);

		byte[] buf = buffer;
		buf[position++] = (byte) (i32 & 0xff);
		buf[position++] = (byte) (i32 >>> 8);
		buf[position++] = (byte) (i32 >>> 16);
		buf[position++] = (byte) (i32 >>> 24);
	}

	/**
	 * Put a string in the buffer.
	 * 
	 * @param s
	 *            the value to put in the buffer
	 */
	protected final void putString(String s) {
		ensureCapacity(position + (s.length() * 2) + 1);

		System.arraycopy(s.getBytes(), 0, buffer, position, s.length());
		position += s.length();
		buffer[position++] = 0;
	}

	protected final void sendBinlogDump(String fileName, final long filePosition, final int serverId,
			boolean nonBlocking) throws IOException {
		position = NET_HEADER_SIZE;

		putByte(COM_BINLOG_DUMP);
		putInt32(filePosition);
		int binlog_flags = nonBlocking ? BINLOG_DUMP_NON_BLOCK : 0;
		binlog_flags |= BINLOG_SEND_ANNOTATE_ROWS_EVENT;
		putInt16(binlog_flags); // binlog_flags
		putInt32(serverId); // slave's server-id
		putString(fileName);

		final byte[] buf = buffer;
		final int len = position - NET_HEADER_SIZE;
		buf[0] = (byte) (len & 0xff);
		buf[1] = (byte) (len >>> 8);
		buf[2] = (byte) (len >>> 16);

		mysqlOutput.write(buffer, 0, position);
		mysqlOutput.flush();
	}

	protected final void sendBinlogDumpGtid(String gtidStr, String fileName, final long filePosition,
			final int serverId, boolean nonBlocking) throws IOException {
		if (gtidStr == null || gtidStr.length() == 0) {
			throw new IllegalArgumentException("gtidstr must set");
		}
		AssitWrite assit = new AssitWrite(1000);
		assit.write((byte) 0x1e, 1);
		assit.write(0, 2);// flag
		assit.write(serverId, 4);
		assit.write(fileName.length(), 4);
		assit.write(fileName);
		assit.write(filePosition, 8);
		GtidSet gtidSet = new GtidSet(gtidStr);
		Collection<GtidSet.UUIDSet> uuidSets = gtidSet.getUUIDSets();
		int dataSize = 8 /* number of uuidSets */;
		for (GtidSet.UUIDSet uuidSet : uuidSets) {
			dataSize += 16 /* uuid */ + 8 /* number of intervals */ + uuidSet.getIntervals().size()
					/* number of intervals */ * 16 /* start-end */;
		}
		assit.write(dataSize, 4);
		assit.write((long) uuidSets.size(), 8);

		for (GtidSet.UUIDSet uuidSet : uuidSets) {
			// assit.write(uuidSet.getUUID().replace("-", ""));
			assit.write(hexToByteArray(uuidSet.getUUID().replace("-", "")));
			Collection<GtidSet.Interval> intervals = uuidSet.getIntervals();
			assit.write((long) intervals.size(), 8);
			for (GtidSet.Interval interval : intervals) {
				assit.write((long) interval.getStart(), 8);
				assit.write((long) interval.getEnd() + 1 /* right-open */, 8);
			}
		}
		byte[] data = assit.get();
		mysqlOutput.write(packBody(data, 0));
		mysqlOutput.flush();
	}

	private final byte[] packBody(byte[] data, int sequence) throws IOException {
		// packetBean.setBody(new LogBuffer(data, 0, data.length));
		// packetBean.setSequence(sequence);

		AssitWrite assit = new AssitWrite(data.length + 4 * (data.length / MAX_PACKET_LENGTH + 1));
		if (data.length < MAX_PACKET_LENGTH) { // Single packet
			assit.write(data.length, 3);
			assit.write(sequence, 1);
			assit.write(data);
			// assit.write(packetBean.getBody().getData());
		} else {
			// If the length of the packet is greater than the value of
			// MAX_PACKET_LENGTH,
			// which is defined to be power(2, 24) �C 1 in sql/net_serv.cc, the
			// packet gets
			// split into smaller packets with bodies of MAX_PACKET_LENGTH plus
			// the
			// last
			// packet with a body that is shorter than MAX_PACKET_LENGTH.
			byte[][] muliBody = AssitWrite.splitBytes(data, MAX_PACKET_LENGTH);
			for (int i = 0; i < muliBody.length; i++) {
				assit.write(MAX_PACKET_LENGTH, 3);
				assit.write(sequence++, 1);
				assit.write(muliBody[i]);
			}
			// The last short packet will always be present even if it must have
			// a
			// zero-length body.
			// It serves as an indicator that there are no more packet parts
			// left in
			// the stream for this large packet.
			assit.write(0, 3);
			assit.write(sequence++, 1);
			assit.write(new byte[0]);
		}
		return assit.get();

	}

	private static byte[] hexToByteArray(String uuid) {
		byte[] b = new byte[uuid.length() / 2];
		for (int i = 0, j = 0; j < uuid.length(); j += 2) {
			b[i++] = (byte) Integer.parseInt(uuid.charAt(j) + "" + uuid.charAt(j + 1), 16);
		}
		return b;
	}

	/**
	 * {@inheritDoc}
	 * 
	 * @see net.wicp.tams.common.binlog.parser.LogFetcher#fetch()
	 */
	public boolean fetch() throws IOException {
		try {
			// Fetching packet header from input.
			if (!fetch0(0, NET_HEADER_SIZE)) {
				logger.warn("Reached end of input stream while fetching header");
				return false;
			}

			// Fetching the first packet(may a multi-packet).
			int netlen = getUint24(PACKET_LEN_OFFSET);
			int netnum = getUint8(PACKET_SEQ_OFFSET);
			if (!fetch0(NET_HEADER_SIZE, netlen)) {
				logger.warn("Reached end of input stream: packet #" + netnum + ", len = " + netlen);
				return false;
			}

			// Detecting error code.
			final int mark = getUint8(NET_HEADER_SIZE);
			if (mark != 0) {
				if (mark == 255) // error from master
				{
					// Indicates an error, for example trying to fetch from
					// wrong
					// binlog position.
					position = NET_HEADER_SIZE + 1;
					final int errno = getInt16();
					String sqlstate = forward(1).getFixString(SQLSTATE_LENGTH);
					String errmsg = getFixString(limit - position);
					throw new IOException("Received error packet:" + " errno = " + errno + ", sqlstate = " + sqlstate
							+ " errmsg = " + errmsg);
				} else if (mark == 254) {
					// Indicates end of stream. It's not clear when this would
					// be sent.
					logger.warn("Received EOF packet from server, apparent" + " master disconnected.");
					return false;
				} else {
					// Should not happen.
					throw new IOException("Unexpected response " + mark + " while fetching binlog: packet #" + netnum
							+ ", len = " + netlen);
				}
			}

			// The first packet is a multi-packet, concatenate the packets.
			while (netlen == MAX_PACKET_LENGTH) {
				if (!fetch0(0, NET_HEADER_SIZE)) {
					logger.warn("Reached end of input stream while fetching header");
					return false;
				}

				netlen = getUint24(PACKET_LEN_OFFSET);
				netnum = getUint8(PACKET_SEQ_OFFSET);
				if (!fetch0(limit, netlen)) {
					logger.warn("Reached end of input stream: packet #" + netnum + ", len = " + netlen);
					return false;
				}
			}

			// Preparing buffer variables to decoding.
			origin = NET_HEADER_SIZE + 1;
			position = origin;
			limit -= origin;
			return true;
		} catch (SocketTimeoutException e) {
			close(); /* Do cleanup */
			logger.error("Socket timeout expired, closing connection", e);
			throw e;
		} catch (InterruptedIOException e) {
			close(); /* Do cleanup */
			logger.warn("I/O interrupted while reading from client socket", e);
			throw e;
		} catch (IOException e) {
			close(); /* Do cleanup */
			logger.error("I/O error while reading from client socket", e);
			throw e;
		}
	}

	private final boolean fetch0(final int off, final int len) throws IOException {
		ensureCapacity(off + len);

		for (int count, n = 0; n < len; n += count) {
			if (0 > (count = mysqlInput.read(buffer, off + n, len - n))) {
				// Reached end of input stream
				return false;
			}
		}

		if (limit < off + len)
			limit = off + len;
		return true;
	}

	/**
	 * {@inheritDoc}
	 * 
	 * @see net.wicp.tams.common.binlog.parser.LogFetcher#close()
	 */
	public void close() throws IOException {
		try {
			if (conn != null)
				conn.close();

			conn = null;
			mysqlInput = null;
			mysqlOutput = null;
		} catch (SQLException e) {
			logger.warn("Unable to close connection", e);
		}
	}
}
